package com.shujia.flink.async

import java.sql.Connection

import com.shujia.flink.async.Demo1Sync.Score
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._

object Demo1Sync {
  def main(args: Array[String]): Unit = {

    /**
      * 使用同步的方式关联维表
      *
      */


    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val scores = env.socketTextStream("node1", 8888)
      .map(line => line.split(","))
      .filter(_.length == 3)
      .map(s => Score(s(0), s(1), s(2).toDouble))

    //关联学生表
    val joinDS = scores.map(new SyncJoinMysql)
    joinDS.print()

    env.execute()

  }

  case class Score(stuId: String, couId: String, sco: Double)

}

class SyncJoinMysql extends RichMapFunction[Score, String] {

  var con: Connection = _

  // 一个分区调用一次
  override def open(parameters: Configuration): Unit = {
    //建立mysql连接
    con = JDBCUtil.getConnection
  }

  //数据按照顺序一条一条处理
  override def map(value: Score): String = {
    //通过学号查询学生信息
    val stat = con.prepareStatement("select * from student where id=?")
    stat.setString(1, value.stuId)

    //执行sql需要时间
    val resultSet = stat.executeQuery()

    var line = value.stuId + "\t" + value.couId + "\t" + value.sco

    //模型数据库访问延迟
    Thread.sleep(1000)

    if (resultSet.next()) {
      val name = resultSet.getString("name")
      val age = resultSet.getInt("age")
      val gender = resultSet.getString("gender")
      val clazz = resultSet.getString("clazz")

      line = line + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz
    }
    line
  }
}
